package com.mjf.spark.day05

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 闭包情况演示
 *
 *   -Spark2.0开始支持另外一种Kryo序列化机制。
 *   -Kryo速度是Serializable的10倍。
 *   -当RDD在Shuffle数据的时候，简单数据类型、数组和字符串类型已经在Spark内部使用kryo来序列化。
 */
object Spark03_TestClosureClean {

  def main(args: Array[String]): Unit = {

    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().
      setAppName("Spark03_TestClosureClean").
      setMaster("local[*]")
      // 替换默认的序列化机制
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      // 注册需要使用 kryo 序列化的自定义类
      .registerKryoClasses(Array(classOf[Search]))


    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3.创建一个RDD
    val rdd: RDD[String] = sc.makeRDD(Array("hello world", "hello spark", "hive", "lucy"))

    //3.1创建一个Search对象
    val search = new Search("hello")

    // Driver：算子以外的代码都是在Driver端执行
    // Executor：算子里面的代码都是在Executor端执行
    //3.2 函数传递，打印：ERROR Task not serializable
    search.getMatch1(rdd).collect().foreach(println)

    //3.3 属性传递，打印：ERROR Task not serializable
    search.getMatche2(rdd).collect().foreach(println)

    //4.关闭连接
    sc.stop()
  }
}

// 样例类默认是序列化的
// case class Search(query: String) {
class Search(query: String){// extends Serializable {

  def isMatch(s: String): Boolean = {
    s.contains(query)
  }

  // 函数序列化案例
  def getMatch1(rdd: RDD[String]): RDD[String] = {
    // 在这个方法中所调用的方法isMatch()是定义在Search这个类中的
    // 实际上调用的是this. isMatch()，this表示Search这个类的对象
    // 程序在运行过程中需要将Search对象序列化以后传递到Executor端
//    rdd.filter(this.isMatch)
    // 解决方案：1.类继承scala.Serializable 2.使用样例类
    rdd.filter(isMatch)
  }

  // 属性序列化案例
  def getMatche2(rdd: RDD[String]): RDD[String] = {
    //在这个方法中所调用的方法query是定义在Search这个类中的字段
    // 实际上调用的是this. query，this表示Search这个类的对象
    // 程序在运行过程中需要将Search对象序列化以后传递到Executor端。
    //rdd.filter(x => x.contains(this.query))
    rdd.filter(x => x.contains(query))
    // 解决方案：1.类继承scala.Serializable 2.将类变量query赋值给局部变量 3.使用样例类
//    val q = query   //将类变量赋值给局部变量
//    rdd.filter(x => x.contains(q))
  }

}


